/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.spark.sql

import java.nio.file.{Files, Path}
import java.util.{Collections, Properties}
import java.util.concurrent.atomic.AtomicLong

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.util.{Failure, Success, Try}

import com.google.protobuf.util.JsonFormat
import com.google.protobuf.util.JsonFormat.TypeRegistry
import io.grpc.inprocess.InProcessChannelBuilder
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}

import org.apache.spark.connect.proto
import org.apache.spark.connect.proto.StorageLevel
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{functions => fn}
import org.apache.spark.sql.avro.{functions => avroFn}
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.StringEncoder
import org.apache.spark.sql.connect.client.SparkConnectClient
import org.apache.spark.sql.connect.client.util.ConnectFunSuite
import org.apache.spark.sql.connect.client.util.IntegrationTestUtils
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.protobuf.{functions => pbFn}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.CalendarInterval
import org.apache.spark.util.SparkFileUtils

// scalastyle:off
/**
 * Test the plans generated by the client. This serves two purposes:
 *
 *   1. Make sure the generated plan matches our expectations. The generated JSON file can be used
 *      for this during review.
 *   1. Make sure the generated plans are stable. Changes to the generated plans should be rare.
 *      The generated plan is compared to the (previously) generated proto file; the test fails
 *      when they are different.
 *
 * If you need to re-generate the golden files, you need to set the SPARK_GENERATE_GOLDEN_FILES=1
 * environment variable before running this test, e.g.:
 * {{{
 *   SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "connect-client-jvm/testOnly org.apache.spark.sql.PlanGenerationTestSuite"
 * }}}
 *
 * If you need to clean the orphaned golden files, you need to set the
 * SPARK_CLEAN_ORPHANED_GOLDEN_FILES=1 environment variable before running this test, e.g.:
 * {{{
 *   SPARK_CLEAN_ORPHANED_GOLDEN_FILES=1 build/sbt "connect-client-jvm/testOnly org.apache.spark.sql.PlanGenerationTestSuite"
 * }}}
 * Note: not all orphaned golden files should be cleaned, some are reserved for testing backups
 * compatibility.
 *
 * Note that the plan protos are used as the input for the `ProtoToParsedPlanTestSuite` in the
 * `connector/connect/server` module
 */
// scalastyle:on
class PlanGenerationTestSuite
    extends ConnectFunSuite
    with BeforeAndAfterAll
    with BeforeAndAfterEach
    with Logging {

  // Borrowed from SparkFunSuite
  private val regenerateGoldenFiles: Boolean = System.getenv("SPARK_GENERATE_GOLDEN_FILES") == "1"

  private val cleanOrphanedGoldenFiles: Boolean =
    System.getenv("SPARK_CLEAN_ORPHANED_GOLDEN_FILES") == "1"

  protected val queryFilePath: Path = commonResourcePath.resolve("query-tests/queries")

  // A relative path to /connector/connect/server, used by `ProtoToParsedPlanTestSuite` to run
  // with the datasource.
  protected val testDataPath: Path = java.nio.file.Paths.get(
    "../",
    "common",
    "src",
    "test",
    "resources",
    "query-tests",
    "test-data")

  private val registry = TypeRegistry
    .newBuilder()
    .add(proto.ExamplePluginRelation.getDescriptor)
    .add(proto.ExamplePluginExpression.getDescriptor)
    .add(proto.ExamplePluginCommand.getDescriptor)
    .build()

  private val printer = JsonFormat.printer().usingTypeRegistry(registry)

  private var session: SparkSession = _

  override protected def beforeAll(): Unit = {
    super.beforeAll()
    val client = SparkConnectClient(InProcessChannelBuilder.forName("/dev/null").build())
    session =
      new SparkSession(client, cleaner = SparkSession.cleaner, planIdGenerator = new AtomicLong)
  }

  override protected def beforeEach(): Unit = {
    session.resetPlanIdGenerator()
  }

  override protected def afterAll(): Unit = {
    session.close()
    if (cleanOrphanedGoldenFiles) {
      cleanOrphanedGoldenFile()
    }
    super.afterAll()
  }

  private def cleanOrphanedGoldenFile(): Unit = {
    val allTestNames = testNames.map(_.replace(' ', '_'))
    val orphans = SparkFileUtils
      .recursiveList(queryFilePath.toFile)
      .filter(g =>
        g.getAbsolutePath.endsWith(".proto.bin") ||
          g.getAbsolutePath.endsWith(".json"))
      .filter(g =>
        !allTestNames.contains(g.getName.stripSuffix(".proto.bin")) &&
          !allTestNames.contains(g.getName.stripSuffix(".json")))
    orphans.foreach(SparkFileUtils.deleteRecursively)
  }

  private def test(name: String)(f: => Dataset[_]): Unit = super.test(name) {
    val actual = f.plan.getRoot
    val goldenFile = queryFilePath.resolve(name.replace(' ', '_') + ".proto.bin")
    Try(readRelation(goldenFile)) match {
      case Success(expected) if expected == actual =>
      // Ok!
      case Success(_) if regenerateGoldenFiles =>
        logInfo("Rewriting Golden File")
        writeGoldenFile(goldenFile, actual)
      case Success(expected) =>
        fail(s"""
             |Expected and actual plans do not match:
             |
             |=== Expected Plan ===
             |$expected
             |
             |=== Actual Plan ===
             |$actual
             |""".stripMargin)
      case Failure(_) if regenerateGoldenFiles =>
        logInfo("Writing Golden File")
        writeGoldenFile(goldenFile, actual)
      case Failure(_) =>
        fail(
          "No golden file found. Please re-run this test with the " +
            "SPARK_GENERATE_GOLDEN_FILES=1 environment variable set")
    }
  }

  private def readRelation(path: Path): proto.Relation = {
    val input = Files.newInputStream(path)
    try proto.Relation.parseFrom(input)
    finally {
      input.close()
    }
  }

  private def writeGoldenFile(path: Path, relation: proto.Relation): Unit = {
    val output = Files.newOutputStream(path)
    try relation.writeTo(output)
    finally {
      output.close()
    }
    // Write the json file for verification.
    val jsonPath =
      path.getParent.resolve(path.getFileName.toString.stripSuffix(".proto.bin") + ".json")
    val writer = Files.newBufferedWriter(jsonPath)
    try writer.write(printer.print(relation))
    finally {
      writer.close()
    }
  }

  private val urlWithUserAndPass = "jdbc:h2:mem:testdb0;user=testUser;password=testPass"

  private val simpleSchema = new StructType()
    .add("id", "long")
    .add("a", "int")
    .add("b", "double")

  private val simpleSchemaString = simpleSchema.catalogString

  private val otherSchema = new StructType()
    .add("a", "int")
    .add("id", "long")
    .add("payload", "binary")

  private val otherSchemaString = otherSchema.catalogString

  private val complexSchema = simpleSchema
    .add("d", simpleSchema)
    .add("e", "array<int>")
    .add("f", MapType(StringType, simpleSchema))
    .add("g", "string")

  private val complexSchemaString = complexSchema.catalogString

  private val binarySchema = new StructType()
    .add("id", "long")
    .add("bytes", "binary")

  private val binarySchemaString = binarySchema.catalogString

  private val temporalsSchema = new StructType()
    .add("d", "date")
    .add("t", "timestamp")
    .add("s", "string")
    .add("x", "bigint")
    .add(
      "wt",
      new StructType()
        .add("start", "timestamp")
        .add("end", "timestamp"))

  private val temporalsSchemaString = temporalsSchema.catalogString

  private val booleanSchema = new StructType()
    .add("id", "long")
    .add("flag", "boolean")

  private val booleanSchemaString = booleanSchema.catalogString

  private def createLocalRelation(schema: String): DataFrame = session.newDataFrame { builder =>
    // TODO API is not consistent. Now we have two different ways of working with schemas!
    builder.getLocalRelationBuilder.setSchema(schema)
  }

  // A few helper dataframes.
  private def simple: DataFrame = createLocalRelation(simpleSchemaString)
  private def left: DataFrame = simple
  private def right: DataFrame = createLocalRelation(otherSchemaString)
  private def complex = createLocalRelation(complexSchemaString)
  private def binary = createLocalRelation(binarySchemaString)
  private def temporals = createLocalRelation(temporalsSchemaString)
  private def boolean = createLocalRelation(booleanSchemaString)

  /* Spark Session API */
  test("range") {
    session.range(1, 10, 1, 2)
  }

  test("read") {
    session.read
      .format("csv")
      .schema(
        StructType(
          StructField("name", StringType) ::
            StructField("age", IntegerType) ::
            StructField("job", StringType) :: Nil))
      .option("header", "true")
      .options(Map("delimiter" -> ";"))
      .load(testDataPath.resolve("people.csv").toString)
  }

  test("read jdbc") {
    session.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES", new Properties())
  }

  test("read jdbc with partition") {
    session.read.jdbc(urlWithUserAndPass, "TEST.EMP", "THEID", 0, 4, 3, new Properties())
  }

  test("read jdbc with predicates") {
    val parts = Array[String]("THEID < 2", "THEID >= 2")
    session.read.jdbc(urlWithUserAndPass, "TEST.PEOPLE", parts, new Properties())
  }

  test("read json") {
    session.read.json(testDataPath.resolve("people.json").toString)
  }

  test("json from dataset") {
    session.read
      .schema(new StructType().add("c1", StringType).add("c2", IntegerType))
      .option("allowSingleQuotes", "true")
      .json(session.emptyDataset(StringEncoder))
  }

  test("toJSON") {
    complex.toJSON
  }

  test("read csv") {
    session.read.csv(testDataPath.resolve("people.csv").toString)
  }

  test("csv from dataset") {
    session.read
      .schema(new StructType().add("c1", StringType).add("c2", IntegerType))
      .option("header", "true")
      .csv(session.emptyDataset(StringEncoder))
  }

  test("read parquet") {
    session.read.parquet(testDataPath.resolve("users.parquet").toString)
  }

  test("read orc") {
    session.read.orc(testDataPath.resolve("users.orc").toString)
  }

  test("read table") {
    session.read.table("myTable")
  }

  test("table") {
    session.table("myTable")
  }

  test("read text") {
    session.read.text(testDataPath.resolve("people.txt").toString)
  }

  /* Dataset API */
  test("select") {
    simple.select(fn.col("id"))
  }

  test("select typed 1-arg") {
    val encoder = ScalaReflection.encoderFor[(Long, Int)]
    simple.select(fn.struct(fn.col("id"), fn.col("a")).as(encoder))
  }

  test("select typed 2-arg") {
    val encoder = ScalaReflection.encoderFor[(Long, Int)]
    val encoder2 = ScalaReflection.encoderFor[(Double, Double)]
    val col = fn.struct(fn.col("id"), fn.col("a"))
    val col2 = fn.struct(fn.col("a"), fn.col("b"))
    simple.select(col.as(encoder), col2.as(encoder2))
  }

  test("select typed 3-arg") {
    val encoder = ScalaReflection.encoderFor[(Long, Int)]
    val encoder2 = ScalaReflection.encoderFor[(Double, Double)]
    val col = fn.struct(fn.col("id"), fn.col("a"))
    val col2 = fn.struct(fn.col("a"), fn.col("b"))
    simple.select(col.as(encoder), col2.as(encoder2), col.as(encoder))
  }

  test("select typed 4-arg") {
    val encoder = ScalaReflection.encoderFor[(Long, Int)]
    val col = fn.struct(fn.col("id"), fn.col("a"))
    simple.select(col.as(encoder), col.as(encoder), col.as(encoder), col.as(encoder))
  }

  test("select typed 5-arg") {
    val encoder = ScalaReflection.encoderFor[(Long, Int)]
    val encoder2 = ScalaReflection.encoderFor[(Double, Double)]
    val col = fn.struct(fn.col("id"), fn.col("a"))
    val col2 = fn.struct(fn.col("a"), fn.col("b"))
    simple.select(
      col.as(encoder),
      col2.as(encoder2),
      col.as(encoder),
      col.as(encoder),
      col2.as(encoder2))
  }

  test("limit") {
    simple.limit(10)
  }

  test("filter") {
    simple.filter(fn.col("id") === fn.lit(10L))
  }

  test("toDF") {
    simple.toDF("x1", "x2", "x3")
  }

  test("to") {
    simple.to(
      new StructType()
        .add("b", "double")
        .add("id", "int"))
  }

  test("join inner_no_condition") {
    left.join(right)
  }

  test("join inner_using_single_col") {
    left.join(right, "id")
  }

  test("join inner_using_multiple_col_array") {
    left.join(right, Array("id", "a"))
  }

  test("join inner_using_multiple_col_seq") {
    left.join(right, Seq("id", "a"))
  }

  test("join using_single_col") {
    left.join(right, "id", "left_semi")
  }

  test("join using_multiple_col_array") {
    left.join(right, Array("id", "a"), "full_outer")
  }

  test("join using_multiple_col_seq") {
    left.join(right, Seq("id", "a"), "right_outer")
  }

  test("join inner_condition") {
    left.alias("l").join(right.alias("r"), fn.col("l.a") === fn.col("r.a"))
  }

  test("join condition") {
    left.as("l").join(right.as("r"), fn.col("l.id") === fn.col("r.id"), "left_anti")
  }

  test("crossJoin") {
    left.crossJoin(right)
  }

  test("sortWithinPartitions strings") {
    simple.sortWithinPartitions("a", "id")
  }

  test("sortWithinPartitions columns") {
    simple.sortWithinPartitions(fn.col("id"), fn.col("b"))
  }

  test("sort strings") {
    simple.sort("b", "a")
  }

  test("sort columns") {
    simple.sort(fn.col("id"), fn.col("b"))
  }

  test("orderBy strings") {
    simple.sort("b", "id", "a")
  }

  test("orderBy columns") {
    simple.sort(fn.col("id"), fn.col("b"), fn.col("a"))
  }

  test("apply") {
    val stable = simple
    stable.select(stable("a"))
  }

  test("hint") {
    simple.hint("coalesce", 100)
  }

  test("col") {
    val stable = simple
    stable.select(stable.col("id"), stable.col("b"))
  }

  test("colRegex") {
    simple.select(simple.colRegex("`a|id`"))
  }

  test("as string") {
    simple.as("foo")
  }

  test("as symbol") {
    simple.as('bar)
  }
  test("alias string") {
    simple.alias("fooz")
  }

  test("alias symbol") {
    simple.alias("bob")
  }

  test("select strings") {
    simple.select("id", "a")
  }

  test("selectExpr") {
    simple.selectExpr("a + 10 as x", "id % 10 as grp")
  }

  test("filter expr") {
    simple.filter("exp(a) < 10.0")
  }

  test("where column") {
    simple.where(fn.col("id") === fn.lit(1L))
  }

  test("where expr") {
    simple.where("a + id < 1000")
  }

  test("unpivot values") {
    simple.unpivot(
      ids = Array(fn.col("id"), fn.col("a")),
      values = Array(fn.col("b")),
      variableColumnName = "name",
      valueColumnName = "value")
  }

  test("unpivot no_values") {
    simple.unpivot(
      ids = Array(fn.col("id")),
      variableColumnName = "name",
      valueColumnName = "value")
  }

  test("melt values") {
    simple.unpivot(
      ids = Array(fn.col("a")),
      values = Array(fn.col("id")),
      variableColumnName = "name",
      valueColumnName = "value")
  }

  test("melt no_values") {
    simple.melt(
      ids = Array(fn.col("id"), fn.col("a")),
      variableColumnName = "name",
      valueColumnName = "value")
  }

  test("offset") {
    simple.offset(1000)
  }

  test("union") {
    simple.union(simple)
  }

  test("unionAll") {
    simple.union(simple)
  }

  test("unionByName") {
    simple.drop("b").unionByName(right.drop("payload"))
  }

  test("unionByName allowMissingColumns") {
    simple.unionByName(right, allowMissingColumns = true)
  }

  test("intersect") {
    simple.intersect(simple)
  }

  test("intersectAll") {
    simple.intersectAll(simple)
  }

  test("except") {
    simple.except(simple)
  }

  test("exceptAll") {
    simple.exceptAll(simple)
  }

  test("sample fraction_seed") {
    simple.sample(0.43, 9890823L)
  }

  test("sample withReplacement_fraction_seed") {
    simple.sample(withReplacement = true, 0.23, 898L)
  }

  test("withColumn single") {
    simple.withColumn("z", fn.expr("a + 100"))
  }

  test("withColumns scala_map") {
    simple.withColumns(Map(("b", fn.lit("redacted")), ("z", fn.expr("a + 100"))))
  }

  test("withColumns java_map") {
    val map = new java.util.HashMap[String, Column]
    map.put("g", fn.col("id"))
    map.put("a", fn.lit("123"))
    simple.withColumns(map)
  }

  test("withColumnRenamed single") {
    simple.withColumnRenamed("id", "nid")
  }

  test("withColumnRenamed scala_map") {
    simple.withColumnsRenamed(Map(("a", "alpha"), ("b", "beta")))
  }

  test("withColumnRenamed java_map") {
    val map = new java.util.HashMap[String, String]
    map.put("id", "nid")
    map.put("b", "bravo")
    simple.withColumnsRenamed(map)
  }

  test("withMetadata") {
    val builder = new MetadataBuilder
    builder.putString("description", "unique identifier")
    simple.withMetadata("id", builder.build())
  }

  test("drop single string") {
    simple.drop("a")
  }

  test("drop multiple strings") {
    simple.drop("id", "a", "b")
  }

  test("drop single column") {
    simple.drop(fn.col("b"))
  }

  test("drop multiple column") {
    simple.drop(fn.col("b"), fn.col("id"))
  }

  test("dropDuplicates") {
    simple.dropDuplicates()
  }

  test("dropDuplicates names seq") {
    simple.dropDuplicates("a" :: "b" :: Nil)
  }

  test("dropDuplicates names array") {
    simple.dropDuplicates(Array("a", "id"))
  }

  test("dropDuplicates varargs") {
    simple.dropDuplicates("a", "b", "id")
  }

  test("describe") {
    simple.describe("id", "b")
  }

  test("summary") {
    simple.summary("mean", "min")
  }

  test("repartition") {
    simple.repartition(24)
  }

  test("repartition num_partitions_expressions") {
    simple.repartition(22, fn.col("a"), fn.col("id"))
  }

  test("repartition expressions") {
    simple.repartition(fn.col("id"), fn.col("b"))
  }

  test("repartitionByRange num_partitions_expressions") {
    simple.repartitionByRange(33, fn.col("b"), fn.col("id").desc_nulls_first)
  }

  test("repartitionByRange expressions") {
    simple.repartitionByRange(fn.col("a").asc, fn.col("id").desc_nulls_first)
  }

  test("coalesce") {
    simple.coalesce(5)
  }

  test("distinct") {
    simple.distinct()
  }

  /* Column API */
  private def columnTest(name: String)(f: => Column): Unit = {
    test("column " + name) {
      complex.select(f)
    }
  }

  private def orderColumnTest(name: String)(f: => Column): Unit = {
    test("column " + name) {
      complex.orderBy(f)
    }
  }

  columnTest("apply") {
    fn.col("f").apply("super_duper_key")
  }

  columnTest("unary minus") {
    -fn.lit(1)
  }

  columnTest("not") {
    !fn.lit(true)
  }

  columnTest("equals") {
    fn.col("a") === fn.col("b")
  }

  columnTest("not equals") {
    fn.col("a") =!= fn.col("b")
  }

  columnTest("gt") {
    fn.col("a") > fn.col("b")
  }

  columnTest("lt") {
    fn.col("a") < fn.col("b")
  }

  columnTest("geq") {
    fn.col("a") >= fn.col("b")
  }

  columnTest("leq") {
    fn.col("a") <= fn.col("b")
  }

  columnTest("eqNullSafe") {
    fn.col("a") <=> fn.col("b")
  }

  columnTest("when otherwise") {
    val a = fn.col("a")
    fn.when(a < 10, "low").when(a < 20, "medium").otherwise("high")
  }

  columnTest("between") {
    fn.col("a").between(10, 20)
  }

  columnTest("isNaN") {
    fn.col("b").isNaN
  }

  columnTest("isNull") {
    fn.col("g").isNull
  }

  columnTest("isNotNull") {
    fn.col("g").isNotNull
  }

  columnTest("and") {
    fn.col("a") > 10 && fn.col("b") < 0.5d
  }

  columnTest("or") {
    fn.col("a") > 10 || fn.col("b") < 0.5d
  }

  columnTest("add") {
    fn.col("a") + fn.col("b")
  }

  columnTest("subtract") {
    fn.col("a") - fn.col("b")
  }

  columnTest("multiply") {
    fn.col("a") * fn.col("b")
  }

  columnTest("divide") {
    fn.col("a") / fn.col("b")
  }

  columnTest("modulo") {
    fn.col("a") % 10
  }

  columnTest("isin") {
    fn.col("g").isin("hello", "world", "foo")
  }

  columnTest("like") {
    fn.col("g").like("%bob%")
  }

  columnTest("rlike") {
    fn.col("g").like("^[0-9]*$")
  }

  columnTest("ilike") {
    fn.col("g").like("%fOb%")
  }

  columnTest("getItem") {
    fn.col("e").getItem(3)
  }

  columnTest("withField") {
    fn.col("d").withField("x", fn.lit("xq"))
  }

  columnTest("dropFields") {
    fn.col("d").dropFields("a", "c")
  }

  columnTest("getField") {
    fn.col("d").getItem("b")
  }

  columnTest("substr") {
    fn.col("g").substr(8, 3)
  }

  columnTest("contains") {
    fn.col("g").contains("baz")
  }

  columnTest("startsWith") {
    fn.col("g").startsWith("prefix_")
  }

  columnTest("endsWith") {
    fn.col("g").endsWith("suffix_")
  }

  columnTest("alias") {
    fn.col("a").name("b")
  }

  columnTest("as multi") {
    fn.expr("inline(map_values(f))").as(Array("v1", "v2", "v3"))
  }

  columnTest("as with metadata") {
    val builder = new MetadataBuilder
    builder.putString("comment", "modified E field")
    fn.col("e").as("e_mod", builder.build())
  }

  columnTest("cast") {
    fn.col("a").cast("long")
  }

  orderColumnTest("desc") {
    fn.col("b").desc
  }

  orderColumnTest("desc_nulls_first") {
    fn.col("b").desc_nulls_first
  }

  orderColumnTest("desc_nulls_last") {
    fn.col("b").desc_nulls_last
  }

  orderColumnTest("asc") {
    fn.col("a").asc
  }

  orderColumnTest("asc_nulls_first") {
    fn.col("a").asc_nulls_first
  }

  orderColumnTest("asc_nulls_last") {
    fn.col("a").asc_nulls_last
  }

  columnTest("bitwiseOR") {
    fn.col("a").bitwiseOR(7)
  }

  columnTest("bitwiseAND") {
    fn.col("a").bitwiseAND(255)
  }

  columnTest("bitwiseXOR") {
    fn.col("a").bitwiseXOR(78)
  }

  columnTest("star") {
    fn.col("*")
  }

  columnTest("star with target") {
    fn.col("d.*")
  }

  /* Function API */
  private def functionTest(name: String)(f: => Column): Unit = {
    test("function " + name) {
      complex.select(f)
    }
  }

  functionTest("col") {
    fn.col("id")
  }

  functionTest("asc") {
    fn.asc("a")
  }

  functionTest("asc_nulls_first") {
    fn.asc_nulls_first("a")
  }

  functionTest("asc_nulls_last") {
    fn.asc_nulls_last("a")
  }

  functionTest("desc") {
    fn.desc("a")
  }

  functionTest("desc_nulls_first") {
    fn.desc_nulls_first("a")
  }

  functionTest("desc_nulls_last") {
    fn.desc_nulls_last("a")
  }

  functionTest("approx_count_distinct") {
    fn.approx_count_distinct("a")
  }

  functionTest("approx_count_distinct rsd") {
    fn.approx_count_distinct("a", 0.1)
  }

  functionTest("avg") {
    fn.avg("a")
  }

  functionTest("collect_list") {
    fn.collect_list("a")
  }

  functionTest("collect_set") {
    fn.collect_set("a")
  }

  functionTest("corr") {
    fn.corr("a", "b")
  }

  functionTest("count") {
    fn.count(fn.col("a"))
  }

  test("function count typed") {
    simple.select(fn.count("a"))
  }

  functionTest("countDistinct") {
    fn.countDistinct("a", "g")
  }

  functionTest("covar_pop") {
    fn.covar_pop("a", "b")
  }

  functionTest("covar_samp") {
    fn.covar_samp("a", "b")
  }

  functionTest("first with ignore nulls") {
    fn.first("a", ignoreNulls = true)
  }

  functionTest("first with respect nulls") {
    fn.first("a")
  }

  functionTest("first_value with ignore nulls") {
    fn.first_value(fn.col("a"), ignoreNulls = lit(true))
  }

  functionTest("first_value with respect nulls") {
    fn.first_value(fn.col("a"))
  }

  functionTest("any_value with ignore nulls") {
    fn.any_value(fn.col("a"), ignoreNulls = lit(true))
  }

  functionTest("any_value with respect nulls") {
    fn.any_value(fn.col("a"))
  }

  functionTest("kurtosis") {
    fn.kurtosis("a")
  }

  functionTest("last with ignore nulls") {
    fn.last("a", ignoreNulls = true)
  }

  functionTest("last with respect nulls") {
    fn.last("a")
  }

  functionTest("last_value with ignore nulls") {
    fn.last_value(fn.col("a"), ignoreNulls = lit(true))
  }

  functionTest("last_value with respect nulls") {
    fn.last_value(fn.col("a"))
  }

  functionTest("count_if") {
    fn.count_if(fn.col("a").gt(0))
  }

  functionTest("histogram_numeric") {
    fn.histogram_numeric(fn.col("a"), lit(10))
  }

  functionTest("bit_and") {
    fn.bit_and(fn.col("a"))
  }

  functionTest("bit_or") {
    fn.bit_or(fn.col("a"))
  }

  functionTest("bit_xor") {
    fn.bit_xor(fn.col("a"))
  }

  functionTest("mode") {
    fn.mode(fn.col("a"))
  }

  test("function max") {
    simple.select(fn.max("id"))
  }

  functionTest("max_by") {
    fn.max_by(fn.col("a"), fn.col("b"))
  }

  functionTest("median") {
    fn.median(fn.col("a"))
  }

  functionTest("min") {
    fn.min("a")
  }

  functionTest("min_by") {
    fn.min_by(fn.col("a"), fn.col("b"))
  }

  functionTest("percentile without frequency") {
    fn.percentile(fn.col("a"), fn.lit(0.3))
  }

  functionTest("percentile with frequency") {
    fn.percentile(fn.col("a"), fn.lit(0.3), fn.lit(2))
  }

  functionTest("percentile_approx") {
    fn.percentile_approx(fn.col("a"), fn.lit(0.3), fn.lit(20))
  }

  functionTest("approx_percentile") {
    fn.approx_percentile(fn.col("a"), fn.lit(0.3), fn.lit(20))
  }

  functionTest("product") {
    fn.product(fn.col("a"))
  }

  functionTest("skewness") {
    fn.skewness("a")
  }

  functionTest("stddev") {
    fn.stddev("a")
  }

  functionTest("std") {
    fn.std(fn.col("a"))
  }

  functionTest("stddev_samp") {
    fn.stddev_samp("a")
  }

  functionTest("stddev_pop") {
    fn.stddev_pop("a")
  }

  functionTest("sum") {
    fn.sum("a")
  }

  functionTest("sum_distinct") {
    fn.sum_distinct(fn.col("a"))
  }

  functionTest("variance") {
    fn.variance("a")
  }

  functionTest("var_samp") {
    fn.var_samp("a")
  }

  functionTest("var_pop") {
    fn.var_pop("a")
  }

  functionTest("regr_avgx") {
    fn.regr_avgx(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_avgy") {
    fn.regr_avgy(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_count") {
    fn.regr_count(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_intercept") {
    fn.regr_intercept(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_r2") {
    fn.regr_r2(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_slope") {
    fn.regr_slope(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_sxx") {
    fn.regr_sxx(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_sxy") {
    fn.regr_sxy(fn.col("a"), fn.col("b"))
  }

  functionTest("regr_syy") {
    fn.regr_syy(fn.col("a"), fn.col("b"))
  }

  test("function every") {
    boolean.select(fn.every(fn.col("flag")))
  }

  test("function bool_and") {
    boolean.select(fn.bool_and(fn.col("flag")))
  }

  test("function some") {
    boolean.select(fn.some(fn.col("flag")))
  }

  test("function any") {
    boolean.select(fn.any(fn.col("flag")))
  }

  test("function bool_or") {
    boolean.select(fn.bool_or(fn.col("flag")))
  }

  functionTest("array") {
    fn.array("a", "a")
  }

  functionTest("map") {
    fn.map(fn.col("a"), fn.col("g"), lit(22), lit("dummy"))
  }

  functionTest("map_from_arrays") {
    fn.map_from_arrays(fn.array(lit(1), lit(2)), fn.array(lit("one"), lit("two")))
  }

  functionTest("coalesce") {
    fn.coalesce(fn.col("a"), lit(3))
  }

  functionTest("input_file_name") {
    fn.input_file_name()
  }

  functionTest("isnan") {
    fn.isnan(fn.col("b"))
  }

  functionTest("isnull") {
    fn.isnull(fn.col("a"))
  }

  functionTest("monotonically_increasing_id") {
    fn.monotonically_increasing_id()
  }

  functionTest("nanvl") {
    fn.nanvl(lit(Double.NaN), fn.col("a"))
  }

  functionTest("negate") {
    fn.negate(fn.col("a"))
  }

  functionTest("rand with seed") {
    fn.rand(133)
  }

  functionTest("randn with seed") {
    fn.randn(133)
  }

  functionTest("spark_partition_id") {
    fn.spark_partition_id()
  }

  functionTest("sqrt") {
    fn.sqrt("b")
  }

  functionTest("try_add") {
    fn.try_add(fn.col("a"), fn.col("a"))
  }

  functionTest("try_avg") {
    fn.try_avg(fn.col("a"))
  }

  functionTest("try_divide") {
    fn.try_divide(fn.col("a"), fn.col("a"))
  }

  functionTest("try_multiply") {
    fn.try_multiply(fn.col("a"), fn.col("a"))
  }

  functionTest("try_subtract") {
    fn.try_subtract(fn.col("a"), fn.col("a"))
  }

  functionTest("try_sum") {
    fn.try_sum(fn.col("a"))
  }

  functionTest("try_to_timestamp") {
    fn.try_to_timestamp(fn.col("g"), fn.col("g"))
  }

  functionTest("try_to_timestamp without format") {
    fn.try_to_timestamp(fn.col("g"))
  }

  functionTest("try_to_binary") {
    fn.try_to_binary(fn.col("g"), lit("format"))
  }

  functionTest("try_to_binary without format") {
    fn.try_to_binary(fn.col("g"))
  }

  functionTest("try_to_number") {
    fn.try_to_number(fn.col("g"), lit("99,999"))
  }

  functionTest("try_element_at array") {
    fn.try_element_at(fn.col("e"), fn.col("a"))
  }

  functionTest("try_element_at map") {
    fn.try_element_at(fn.col("f"), fn.col("g"))
  }

  functionTest("struct") {
    fn.struct("a", "d")
  }

  functionTest("bitwise_not") {
    fn.bitwise_not(fn.col("a"))
  }

  functionTest("bit_count") {
    fn.bit_count(fn.col("a"))
  }

  functionTest("bit_get") {
    fn.bit_get(fn.col("a"), lit(0))
  }

  functionTest("getbit") {
    fn.getbit(fn.col("a"), lit(0))
  }

  functionTest("expr") {
    fn.expr("a + 1")
  }

  functionTest("abs") {
    fn.abs(fn.col("a"))
  }

  functionTest("acos") {
    fn.acos("b")
  }

  functionTest("acosh") {
    fn.acosh("b")
  }

  functionTest("asin") {
    fn.asin("b")
  }

  functionTest("asinh") {
    fn.asinh("b")
  }

  functionTest("atan") {
    fn.atan("b")
  }

  functionTest("atan2") {
    fn.atan2(fn.col("a").cast("double"), "b")
  }

  functionTest("atanh") {
    fn.atanh("b")
  }

  functionTest("bin") {
    fn.bin("b")
  }

  functionTest("ceil") {
    fn.ceil("b")
  }

  functionTest("ceil scale") {
    fn.ceil(fn.col("b"), lit(2))
  }

  functionTest("ceiling") {
    fn.ceiling(fn.col("b"))
  }

  functionTest("ceiling scale") {
    fn.ceiling(fn.col("b"), lit(2))
  }

  functionTest("conv") {
    fn.conv(fn.col("b"), 10, 16)
  }

  functionTest("cos") {
    fn.cos("b")
  }

  functionTest("cosh") {
    fn.cosh("b")
  }

  functionTest("cot") {
    fn.cot(fn.col("b"))
  }

  functionTest("csc") {
    fn.csc(fn.col("b"))
  }

  functionTest("e") {
    fn.e()
  }

  functionTest("exp") {
    fn.exp("b")
  }

  functionTest("expm1") {
    fn.expm1("b")
  }

  functionTest("factorial") {
    fn.factorial(fn.col("a") % 10)
  }

  functionTest("floor") {
    fn.floor("b")
  }

  functionTest("floor scale") {
    fn.floor(fn.col("b"), lit(2))
  }

  functionTest("greatest") {
    fn.greatest(fn.col("a"), fn.col("d").getItem("a"))
  }

  functionTest("hex") {
    fn.hex(fn.col("a"))
  }

  functionTest("unhex") {
    fn.unhex(fn.col("a"))
  }

  functionTest("hypot") {
    fn.hypot(fn.col("a"), fn.col("b"))
  }

  functionTest("least") {
    fn.least(fn.col("a"), fn.col("d").getItem("a"))
  }

  functionTest("log") {
    fn.log("b")
  }

  functionTest("ln") {
    fn.ln(fn.col("b"))
  }

  functionTest("log with base") {
    fn.log(2, "b")
  }

  functionTest("log10") {
    fn.log10("b")
  }

  functionTest("log1p") {
    fn.log1p("a")
  }

  functionTest("log2") {
    fn.log2("a")
  }

  functionTest("negative") {
    fn.negative(fn.col("a"))
  }

  functionTest("pi") {
    fn.pi()
  }

  functionTest("positive") {
    fn.positive(fn.col("a"))
  }

  functionTest("pow") {
    fn.pow("a", "b")
  }

  functionTest("power") {
    fn.power(fn.col("a"), fn.col("b"))
  }

  functionTest("pmod") {
    fn.pmod(fn.col("a"), fn.lit(10))
  }

  functionTest("rint") {
    fn.rint("b")
  }

  functionTest("round") {
    fn.round(fn.col("b"), 2)
  }

  functionTest("bround") {
    fn.round(fn.col("b"), 2)
  }

  functionTest("sec") {
    fn.sec(fn.col("b"))
  }

  functionTest("shiftleft") {
    fn.shiftleft(fn.col("b"), 2)
  }

  functionTest("shiftright") {
    fn.shiftright(fn.col("b"), 2)
  }

  functionTest("shiftrightunsigned") {
    fn.shiftrightunsigned(fn.col("b"), 2)
  }

  functionTest("signum") {
    fn.signum("b")
  }

  functionTest("sign") {
    fn.sign(fn.col("b"))
  }

  functionTest("sin") {
    fn.sin("b")
  }

  functionTest("sinh") {
    fn.sinh("b")
  }

  functionTest("tan") {
    fn.tan("b")
  }

  functionTest("tanh") {
    fn.tanh("b")
  }

  functionTest("degrees") {
    fn.degrees("b")
  }

  functionTest("radians") {
    fn.radians("b")
  }

  functionTest("current_catalog") {
    fn.current_catalog()
  }

  functionTest("current_database") {
    fn.current_database()
  }

  functionTest("current_schema") {
    fn.current_schema()
  }

  functionTest("current_user") {
    fn.current_user()
  }

  functionTest("user") {
    fn.user()
  }

  functionTest("md5") {
    fn.md5(fn.col("g").cast("binary"))
  }

  functionTest("sha1") {
    fn.sha1(fn.col("g").cast("binary"))
  }

  functionTest("sha2") {
    fn.sha2(fn.col("g").cast("binary"), 512)
  }

  functionTest("crc32") {
    fn.crc32(fn.col("g").cast("binary"))
  }

  functionTest("hash") {
    fn.hash(fn.col("b"), fn.col("id"))
  }

  functionTest("xxhash64") {
    fn.xxhash64(fn.col("id"), fn.col("a"), fn.col("d"), fn.col("g"))
  }

  functionTest("assert_true with message") {
    fn.assert_true(fn.col("id") > 0, lit("id negative!"))
  }

  functionTest("raise_error") {
    fn.raise_error(fn.lit("kaboom"))
  }

  functionTest("ascii") {
    fn.ascii(fn.col("g"))
  }

  functionTest("base64") {
    fn.base64(fn.col("g").cast("binary"))
  }

  functionTest("bit_length") {
    fn.bit_length(fn.col("g"))
  }

  functionTest("concat_ws") {
    fn.concat_ws("-", fn.col("b"), lit("world"), fn.col("id"))
  }

  functionTest("decode") {
    fn.decode(fn.col("g").cast("binary"), "UTF-8")
  }

  functionTest("encode") {
    fn.encode(fn.col("g"), "UTF-8")
  }

  functionTest("format_number") {
    fn.format_number(fn.col("b"), 1)
  }

  functionTest("initcap") {
    fn.initcap(fn.col("g"))
  }

  functionTest("length") {
    fn.length(fn.col("g"))
  }

  functionTest("len") {
    fn.len(fn.col("g"))
  }

  functionTest("lower") {
    fn.lower(fn.col("g"))
  }

  functionTest("levenshtein") {
    fn.levenshtein(fn.col("g"), lit("bob"))
  }

  functionTest("levenshtein with threshold") {
    fn.levenshtein(fn.col("g"), lit("bob"), 2)
  }

  functionTest("locate") {
    fn.locate("jar", fn.col("g"))
  }

  functionTest("locate with pos") {
    fn.locate("jar", fn.col("g"), 10)
  }

  functionTest("lpad") {
    fn.lpad(fn.col("g"), 10, "-")
  }

  test("function lpad binary") {
    binary.select(fn.lpad(fn.col("bytes"), 5, Array(0xc, 0xa, 0xf, 0xe).map(_.toByte)))
  }

  functionTest("ltrim") {
    fn.ltrim(fn.col("g"))
  }

  functionTest("ltrim with pattern") {
    fn.ltrim(fn.col("g"), "xxx")
  }

  functionTest("octet_length") {
    fn.octet_length(fn.col("g"))
  }

  functionTest("rlike") {
    fn.rlike(fn.col("g"), lit("[a-z]+b"))
  }

  functionTest("regexp") {
    fn.regexp(fn.col("g"), lit("[a-z]+b"))
  }

  functionTest("regexp_like") {
    fn.regexp_like(fn.col("g"), lit("[a-z]+b"))
  }

  functionTest("regexp_count") {
    fn.regexp_count(fn.col("g"), lit("\\d+"))
  }

  functionTest("regexp_extract") {
    fn.regexp_extract(fn.col("g"), "(\\d+)-(\\d+)", 1)
  }

  functionTest("regexp_extract_all without regex group index") {
    fn.regexp_extract_all(fn.col("g"), lit("(\\d+)([a-z]+)"))
  }

  functionTest("regexp_extract_all with regex group index") {
    fn.regexp_extract_all(fn.col("g"), lit("(\\d+)([a-z]+)"), lit(1))
  }

  functionTest("regexp_replace") {
    fn.regexp_replace(fn.col("g"), "(\\d+)", "XXX")
  }

  functionTest("regexp_substr") {
    fn.regexp_substr(fn.col("g"), lit("\\d{2}(a|b|m)"))
  }

  functionTest("regexp_instr without regex group index") {
    fn.regexp_instr(fn.col("g"), lit("\\d+(a|b|m)"))
  }

  functionTest("regexp_instr with regex group index") {
    fn.regexp_instr(fn.col("g"), lit("\\d+(a|b|m)"), lit(1))
  }

  functionTest("unbase64") {
    fn.unbase64(fn.col("g"))
  }

  functionTest("rpad") {
    fn.rpad(fn.col("g"), 10, "-")
  }

  test("function rpad binary") {
    binary.select(fn.rpad(fn.col("bytes"), 5, Array(0xb, 0xa, 0xb, 0xe).map(_.toByte)))
  }

  functionTest("rtrim") {
    fn.rtrim(fn.col("g"))
  }

  functionTest("rtrim with pattern") {
    fn.rtrim(fn.col("g"), "yyy")
  }

  functionTest("split") {
    fn.split(fn.col("g"), ";")
  }

  functionTest("split with limit") {
    fn.split(fn.col("g"), ";", 10)
  }

  functionTest("substring") {
    fn.substring(fn.col("g"), 4, 5)
  }

  functionTest("substring_index") {
    fn.substring_index(fn.col("g"), ";", 5)
  }

  functionTest("overlay") {
    fn.overlay(fn.col("b"), lit("foo"), lit(4))
  }

  functionTest("overlay with len") {
    fn.overlay(fn.col("b"), lit("foo"), lit(4), lit("3"))
  }

  functionTest("sentences") {
    fn.sentences(fn.col("g"))
  }

  functionTest("sentences with locale") {
    fn.sentences(fn.col("g"), lit("en"), lit("US"))
  }

  functionTest("translate") {
    fn.translate(fn.col("g"), "foo", "bar")
  }

  functionTest("trim") {
    fn.trim(fn.col("g"))
  }

  functionTest("trim with pattern") {
    fn.trim(fn.col("g"), "---")
  }

  functionTest("upper") {
    fn.upper(fn.col("g"))
  }

  functionTest("years") {
    fn.years(Column("a"))
  }

  functionTest("months") {
    fn.months(Column("a"))
  }

  functionTest("days") {
    fn.days(Column("a"))
  }

  functionTest("hours") {
    fn.hours(Column("a"))
  }

  temporalFunctionTest("convert_timezone with source time zone") {
    fn.convert_timezone(lit("\"Africa/Dakar\""), lit("\"Asia/Urumqi\""), fn.col("t"))
  }

  temporalFunctionTest("convert_timezone without source time zone") {
    fn.convert_timezone(lit("\"Asia/Urumqi\""), fn.col("t"))
  }

  functionTest("make_dt_interval days hours mins secs") {
    fn.make_dt_interval(fn.col("a"), fn.col("a"), fn.col("a"), fn.col("b"))
  }

  functionTest("make_dt_interval days hours mins") {
    fn.make_dt_interval(fn.col("a"), fn.col("a"), fn.col("a"))
  }

  functionTest("make_dt_interval days hours") {
    fn.make_dt_interval(fn.col("a"), fn.col("a"))
  }

  functionTest("make_dt_interval days") {
    fn.make_dt_interval(fn.col("a"))
  }

  functionTest("make_dt_interval") {
    fn.make_dt_interval()
  }

  functionTest("make_interval years months weeks days hours mins secs") {
    fn.make_interval(
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("b"))
  }

  functionTest("make_interval years months weeks days hours mins") {
    fn.make_interval(fn.col("a"), fn.col("a"), fn.col("a"), fn.col("a"), fn.col("a"), fn.col("a"))
  }

  functionTest("make_interval years months weeks days hours") {
    fn.make_interval(fn.col("a"), fn.col("a"), fn.col("a"), fn.col("a"), fn.col("a"))
  }

  functionTest("make_interval years months weeks days") {
    fn.make_interval(fn.col("a"), fn.col("a"), fn.col("a"), fn.col("a"))
  }

  functionTest("make_interval years months weeks") {
    fn.make_interval(fn.col("a"), fn.col("a"), fn.col("a"))
  }

  functionTest("make_interval years months") {
    fn.make_interval(fn.col("a"), fn.col("a"))
  }

  functionTest("make_interval years") {
    fn.make_interval(fn.col("a"))
  }

  functionTest("make_interval") {
    fn.make_interval()
  }

  functionTest("make_timestamp with timezone") {
    fn.make_timestamp(
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("b"),
      fn.col("g"))
  }

  functionTest("make_timestamp without timezone") {
    fn.make_timestamp(
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("b"))
  }

  functionTest("make_timestamp_ltz with timezone") {
    fn.make_timestamp_ltz(
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("b"),
      fn.col("g"))
  }

  functionTest("make_timestamp_ltz without timezone") {
    fn.make_timestamp_ltz(
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("b"))
  }

  functionTest("make_timestamp_ntz") {
    fn.make_timestamp_ntz(
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("a"),
      fn.col("b"))
  }

  functionTest("make_ym_interval years months") {
    fn.make_ym_interval(fn.col("a"), fn.col("a"))
  }

  functionTest("make_ym_interval years") {
    fn.make_ym_interval(fn.col("a"))
  }

  functionTest("make_ym_interval") {
    fn.make_ym_interval()
  }

  functionTest("bucket") {
    fn.bucket(3, Column("a"))
  }

  functionTest("cume_dist") {
    fn.cume_dist().over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("dense_rank") {
    fn.dense_rank().over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("lag") {
    fn.lag(Column("g"), 1, null, ignoreNulls = true)
      .over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("lead") {
    fn.lead(Column("g"), 2, "dv", ignoreNulls = true)
      .over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("nth_value") {
    fn.nth_value(Column("g"), 3, ignoreNulls = true)
      .over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("ntile") {
    fn.ntile(4).over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("percent_rank") {
    fn.percent_rank().over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("rank") {
    fn.rank().over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("row_number") {
    fn.row_number().over(Window.partitionBy(Column("a")).orderBy(Column("id")))
  }

  functionTest("bitmap_bucket_number") {
    fn.bitmap_bit_position(fn.col("id"))
  }

  functionTest("bitmap_bit_position") {
    fn.bitmap_bit_position(fn.col("id"))
  }

  functionTest("bitmap_construct_agg") {
    fn.bitmap_construct_agg(fn.col("id"))
  }

  test("function bitmap_count") {
    binary.select(fn.bitmap_count(fn.col("bytes")))
  }

  test("function bitmap_or_agg") {
    binary.select(fn.bitmap_or_agg(fn.col("bytes")))
  }

  private def temporalFunctionTest(name: String)(f: => Column): Unit = {
    test("function " + name) {
      temporals.select(f)
    }
  }

  temporalFunctionTest("add_months") {
    fn.add_months(fn.col("d"), 2)
  }

  temporalFunctionTest("curdate") {
    fn.curdate()
  }

  temporalFunctionTest("current_date") {
    fn.current_date()
  }

  temporalFunctionTest("current_timezone") {
    fn.current_timezone()
  }

  temporalFunctionTest("current_timestamp") {
    fn.current_timestamp()
  }

  temporalFunctionTest("now") {
    fn.now()
  }

  temporalFunctionTest("localtimestamp") {
    fn.localtimestamp()
  }

  temporalFunctionTest("date_format") {
    fn.date_format(fn.col("d"), "yyyy-MM-dd")
  }

  temporalFunctionTest("date_add") {
    fn.date_add(fn.col("d"), 2)
  }

  temporalFunctionTest("dateadd") {
    fn.dateadd(fn.col("d"), lit(2))
  }

  temporalFunctionTest("date_sub") {
    fn.date_sub(fn.col("d"), 2)
  }

  temporalFunctionTest("datediff") {
    fn.datediff(fn.col("d"), fn.make_date(lit(2020), lit(10), lit(10)))
  }

  temporalFunctionTest("date_diff") {
    fn.date_diff(fn.col("d"), fn.make_date(lit(2020), lit(10), lit(10)))
  }

  temporalFunctionTest("date_from_unix_date") {
    fn.date_from_unix_date(lit(10))
  }

  temporalFunctionTest("year") {
    fn.year(fn.col("d"))
  }

  temporalFunctionTest("quarter") {
    fn.quarter(fn.col("d"))
  }

  temporalFunctionTest("month") {
    fn.month(fn.col("d"))
  }

  temporalFunctionTest("dayofweek") {
    fn.dayofweek(fn.col("d"))
  }

  temporalFunctionTest("dayofmonth") {
    fn.dayofmonth(fn.col("d"))
  }

  temporalFunctionTest("day") {
    fn.day(fn.col("d"))
  }

  temporalFunctionTest("dayofyear") {
    fn.dayofyear(fn.col("d"))
  }

  temporalFunctionTest("hour") {
    fn.hour(fn.col("t"))
  }

  temporalFunctionTest("last_day") {
    fn.last_day(fn.col("t"))
  }

  temporalFunctionTest("minute") {
    fn.minute(fn.col("t"))
  }

  temporalFunctionTest("weekday") {
    fn.weekday(fn.col("d"))
  }

  temporalFunctionTest("make_date") {
    fn.make_date(fn.lit(2018), fn.lit(5), fn.lit(14))
  }

  temporalFunctionTest("months_between") {
    fn.months_between(fn.current_date(), fn.col("d"))
  }

  temporalFunctionTest("months_between with roundoff") {
    fn.months_between(fn.current_date(), fn.col("d"), roundOff = true)
  }

  temporalFunctionTest("next_day") {
    fn.next_day(fn.col("d"), "Mon")
  }

  temporalFunctionTest("second") {
    fn.second(fn.col("t"))
  }

  temporalFunctionTest("weekofyear") {
    fn.weekofyear(fn.col("d"))
  }

  temporalFunctionTest("extract") {
    fn.extract(lit("year"), fn.col("d"))
  }

  temporalFunctionTest("date_part") {
    fn.date_part(lit("year"), fn.col("d"))
  }

  temporalFunctionTest("datepart") {
    fn.datepart(lit("year"), fn.col("d"))
  }

  temporalFunctionTest("from_unixtime") {
    fn.from_unixtime(lit(1L))
  }

  temporalFunctionTest("unix_timestamp") {
    fn.unix_timestamp()
  }

  temporalFunctionTest("unix_timestamp with format") {
    fn.unix_timestamp(fn.col("s"), "yyyy-MM-dd HH:mm:ss.SSSS")
  }

  temporalFunctionTest("to_timestamp") {
    fn.to_timestamp(fn.col("s"))
  }

  temporalFunctionTest("to_timestamp with format") {
    fn.to_timestamp(fn.col("s"), "yyyy-MM-dd HH:mm:ss.SSSS")
  }

  temporalFunctionTest("to_date") {
    fn.to_date(fn.col("s"))
  }

  temporalFunctionTest("to_date with format") {
    fn.to_date(fn.col("s"), "yyyy-MM-dd")
  }

  temporalFunctionTest("xpath") {
    fn.xpath(fn.col("s"), lit("a/b/text()"))
  }

  temporalFunctionTest("xpath_boolean") {
    fn.xpath_boolean(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_double") {
    fn.xpath_double(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_number") {
    fn.xpath_number(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_float") {
    fn.xpath_float(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_int") {
    fn.xpath_int(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_long") {
    fn.xpath_long(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_short") {
    fn.xpath_short(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("xpath_string") {
    fn.xpath_string(fn.col("s"), lit("a/b"))
  }

  temporalFunctionTest("unix_date") {
    fn.unix_date(fn.to_date(fn.col("s"), "yyyy-MM-dd"))
  }

  temporalFunctionTest("unix_seconds") {
    fn.unix_seconds(fn.to_timestamp(fn.col("s"), "yyyy-MM-dd HH:mm:ss.SSSS"))
  }

  temporalFunctionTest("unix_millis") {
    fn.unix_millis(fn.to_timestamp(fn.col("s"), "yyyy-MM-dd HH:mm:ss.SSSS"))
  }

  temporalFunctionTest("unix_micros") {
    fn.unix_micros(fn.to_timestamp(fn.col("s"), "yyyy-MM-dd HH:mm:ss.SSSS"))
  }

  temporalFunctionTest("trunc") {
    fn.trunc(fn.col("d"), "mm")
  }

  temporalFunctionTest("date_trunc") {
    fn.trunc(fn.col("t"), "minute")
  }

  temporalFunctionTest("from_utc_timestamp") {
    fn.from_utc_timestamp(fn.col("t"), "-08:00")
  }

  temporalFunctionTest("to_utc_timestamp") {
    fn.to_utc_timestamp(fn.col("t"), "-04:00")
  }

  temporalFunctionTest("window") {
    fn.window(fn.col("t"), "1 second")
  }

  test("function window_time") {
    val metadata = new MetadataBuilder().putBoolean("spark.timeWindow", value = true).build()
    temporals
      .withMetadata("wt", metadata)
      .select(fn.window_time(fn.col("wt")))
  }

  temporalFunctionTest("session_window") {
    fn.session_window(fn.col("t"), "10 minutes")
  }

  temporalFunctionTest("timestamp_seconds") {
    fn.timestamp_seconds(fn.col("x"))
  }

  temporalFunctionTest("timestamp_millis") {
    fn.timestamp_millis(fn.col("x"))
  }

  temporalFunctionTest("timestamp_micros") {
    fn.timestamp_micros(fn.col("x"))
  }

  // Array of Long
  // Array of Long
  // Array of Array of Long
  // Map string, Long
  // Map string, Long

  functionTest("array_contains") {
    fn.array_contains(fn.col("e"), lit(1))
  }

  functionTest("array_append") {
    fn.array_append(fn.col("e"), lit(1))
  }

  functionTest("arrays_overlap") {
    fn.arrays_overlap(fn.col("e"), fn.array(lit(1), lit(2)))
  }

  functionTest("slice") {
    fn.slice(fn.col("e"), 0, 5)
  }

  functionTest("array_join") {
    fn.array_join(fn.col("e"), ";")
  }

  functionTest("array_join with null replacement") {
    fn.array_join(fn.col("e"), ";", "null")
  }

  functionTest("concat") {
    fn.concat(fn.col("e"), fn.array(lit(1), lit(2)), fn.sequence(lit(33), lit(40)))
  }

  functionTest("array_position") {
    fn.array_position(fn.col("e"), 10)
  }

  functionTest("element_at") {
    fn.element_at(fn.col("f"), "bob")
  }

  functionTest("get") {
    fn.get(fn.col("e"), lit(2))
  }

  functionTest("array_sort") {
    fn.array_sort(fn.col("e"))
  }

  functionTest("array_sort with comparator") {
    fn.array_sort(fn.col("e"), (l, r) => l - r)
  }

  functionTest("array_remove") {
    fn.array_remove(fn.col("e"), 314)
  }

  functionTest("array_compact") {
    fn.array_compact(fn.col("e"))
  }

  functionTest("array_distinct") {
    fn.array_distinct(fn.col("e"))
  }

  functionTest("array_prepend") {
    fn.array_prepend(fn.col("e"), lit(1))
  }

  functionTest("array_intersect") {
    fn.array_intersect(fn.col("e"), fn.array(lit(10), lit(4)))
  }

  functionTest("array_insert") {
    fn.array_insert(fn.col("e"), lit(0), lit(1))
  }

  functionTest("array_union") {
    fn.array_union(fn.col("e"), fn.array(lit(1), lit(2), lit(3)))
  }

  functionTest("array_except") {
    fn.array_except(fn.col("e"), fn.array(lit(1), lit(2), lit(4)))
  }

  functionTest("transform") {
    fn.transform(fn.col("e"), x => x + 1)
  }

  functionTest("transform with index") {
    fn.transform(fn.col("e"), (x, i) => x + i)
  }

  functionTest("exists") {
    fn.exists(fn.col("e"), x => x > 10)
  }

  functionTest("forall") {
    fn.forall(fn.col("e"), x => x > 10)
  }

  functionTest("filter") {
    fn.filter(fn.col("e"), x => x > 10)
  }

  functionTest("filter with pair input") {
    fn.filter(fn.col("e"), (x, i) => x > 10 && i > 2)
  }

  functionTest("aggregate") {
    fn.aggregate(fn.col("e"), lit(0), (x, y) => x + y)
  }

  functionTest("reduce") {
    fn.reduce(fn.col("e"), lit(0), (x, y) => x + y)
  }

  functionTest("zip_with") {
    fn.zip_with(fn.col("e"), fn.col("e"), (x, y) => x + y)
  }

  functionTest("transform_keys") {
    fn.transform_keys(fn.col("f"), (k, v) => fn.concat(k, v.getItem("id")))
  }

  functionTest("transform_values") {
    fn.transform_values(fn.col("f"), (k, v) => v.withField("key", k))
  }

  functionTest("map_filter") {
    fn.map_filter(fn.col("f"), (k, _) => k.contains(lit("baz")))
  }

  functionTest("map_zip_with") {
    fn.map_zip_with(fn.col("f"), fn.col("f"), (_, v1, v2) => v1.getItem("id") + v2.getItem("id"))
  }

  functionTest("explode") {
    fn.explode(fn.col("e"))
  }

  functionTest("explode_outer") {
    fn.explode_outer(fn.col("e"))
  }

  functionTest("posexplode") {
    fn.posexplode(fn.col("e"))
  }

  functionTest("posexplode_outer") {
    fn.posexplode_outer(fn.col("e"))
  }

  functionTest("inline") {
    fn.inline(fn.map_values(fn.col("f")))
  }

  functionTest("inline_outer") {
    fn.inline_outer(fn.map_values(fn.col("f")))
  }

  functionTest("get_json_object") {
    fn.get_json_object(fn.col("g"), "$.device_type")
  }

  functionTest("json_tuple") {
    fn.json_tuple(fn.col("g"), "a", "b", "id")
  }

  functionTest("from_json") {
    fn.from_json(fn.col("g"), simpleSchema)
  }

  functionTest("schema_of_json") {
    fn.schema_of_json(lit("""[{"col":01}]"""))
  }

  functionTest("schema_of_json with options") {
    fn.schema_of_json(
      lit("""[{"col":01}]"""),
      Collections.singletonMap("allowNumericLeadingZeros", "true"))
  }

  functionTest("to_json") {
    fn.to_json(fn.col("d"), Map(("timestampFormat", "dd/MM/yyyy")))
  }

  functionTest("size") {
    fn.size(fn.col("f"))
  }

  functionTest("sort_array") {
    fn.sort_array(fn.col("e"))
  }

  functionTest("array_min") {
    fn.array_min(fn.col("e"))
  }

  functionTest("array_max") {
    fn.array_max(fn.col("e"))
  }

  functionTest("reverse") {
    fn.reverse(fn.col("e"))
  }

  functionTest("flatten") {
    fn.flatten(fn.array(fn.col("e"), fn.sequence(fn.lit(1), fn.lit(10))))
  }

  functionTest("sequence") {
    fn.sequence(fn.lit(1), fn.lit(10))
  }

  functionTest("array_repeat") {
    fn.array_repeat(fn.col("a"), 10)
  }

  functionTest("map_contains_key") {
    fn.map_contains_key(fn.col("f"), "xyz")
  }

  functionTest("map_keys") {
    fn.map_keys(fn.col("f"))
  }

  functionTest("map_values") {
    fn.map_values(fn.col("f"))
  }

  functionTest("map_entries") {
    fn.map_entries(fn.col("f"))
  }

  functionTest("map_from_entries") {
    fn.map_from_entries(fn.transform(fn.col("e"), (x, i) => fn.struct(i, x)))
  }

  functionTest("arrays_zip") {
    fn.arrays_zip(fn.col("e"), fn.sequence(lit(1), lit(20)))
  }

  functionTest("map_concat") {
    fn.map_concat(
      fn.col("f"),
      fn.map(lit("foo"), fn.struct(lit(12L).as("id"), lit(68).as("a"), lit(Math.E).as("b"))))
  }

  functionTest("from_csv") {
    fn.from_csv(fn.col("g"), simpleSchema, Map(("mode", "FAILFAST")))
  }

  functionTest("schema_of_csv") {
    fn.schema_of_csv(lit("1|abc"), Collections.singletonMap("sep", "|"))
  }

  functionTest("to_csv") {
    fn.to_csv(fn.col("d"), Collections.singletonMap("sep", "|"))
  }

  functionTest("str_to_map") {
    fn.str_to_map(fn.col("g"))
  }

  functionTest("str_to_map with pair delimiter") {
    fn.str_to_map(fn.col("g"), lit(","), lit("="))
  }

  functionTest("str_to_map with pair and keyValue delimiter") {
    fn.str_to_map(fn.col("g"), lit(","))
  }

  functionTest("to_binary") {
    fn.to_binary(fn.col("g"))
  }

  functionTest("to_binary with format") {
    fn.to_binary(fn.col("g"), lit("utf-8"))
  }

  functionTest("to_char") {
    fn.to_char(fn.col("b"), lit("$99.99"))
  }

  functionTest("to_varchar") {
    fn.to_varchar(fn.col("b"), lit("$99.99"))
  }

  functionTest("to_number") {
    fn.to_number(fn.col("g"), lit("$99.99"))
  }

  functionTest("replace") {
    fn.replace(fn.col("g"), fn.col("g"))
  }

  functionTest("replace with specified string") {
    fn.replace(fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("split_part") {
    fn.split_part(fn.col("g"), fn.col("g"), fn.col("a"))
  }

  functionTest("substr") {
    fn.substr(fn.col("g"), fn.col("a"))
  }

  functionTest("substr with len") {
    fn.substr(fn.col("g"), fn.col("a"), fn.col("a"))
  }

  functionTest("parse_url") {
    fn.parse_url(fn.col("g"), fn.col("g"))
  }

  functionTest("parse_url with key") {
    fn.parse_url(fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("printf") {
    fn.printf(fn.col("g"), fn.col("a"), fn.col("g"))
  }

  functionTest("url_decode") {
    fn.url_decode(fn.col("g"))
  }

  functionTest("url_encode") {
    fn.url_encode(fn.col("g"))
  }

  functionTest("position") {
    fn.position(fn.col("g"), fn.col("g"))
  }

  functionTest("position with start") {
    fn.position(fn.col("g"), fn.col("g"), fn.col("a"))
  }

  functionTest("endswith") {
    fn.endswith(fn.col("g"), fn.col("g"))
  }

  functionTest("startswith") {
    fn.startswith(fn.col("g"), fn.col("g"))
  }

  functionTest("to_timestamp_ltz") {
    fn.to_timestamp_ltz(fn.col("g"))
  }

  functionTest("to_timestamp_ltz with format") {
    fn.to_timestamp_ltz(fn.col("g"), fn.col("g"))
  }

  functionTest("to_timestamp_ntz") {
    fn.to_timestamp_ntz(fn.col("g"))
  }

  functionTest("to_timestamp_ntz with format") {
    fn.to_timestamp_ntz(fn.col("g"), fn.col("g"))
  }

  functionTest("to_unix_timestamp") {
    fn.to_unix_timestamp(fn.col("g"))
  }

  functionTest("to_unix_timestamp with format") {
    fn.to_unix_timestamp(fn.col("g"), fn.col("g"))
  }

  functionTest("ifnull") {
    fn.ifnull(fn.col("g"), fn.col("g"))
  }

  functionTest("isnotnull") {
    fn.isnotnull(fn.col("g"))
  }

  functionTest("equal_null") {
    fn.equal_null(fn.col("g"), fn.col("g"))
  }

  functionTest("nullif") {
    fn.nullif(fn.col("g"), fn.col("g"))
  }

  functionTest("nvl") {
    fn.nvl(fn.col("g"), fn.col("g"))
  }

  functionTest("nvl2") {
    fn.nvl2(fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("char") {
    fn.char(fn.col("a"))
  }

  functionTest("btrim") {
    fn.btrim(fn.col("g"))
  }

  functionTest("btrim with specified trim string") {
    fn.btrim(fn.col("g"), fn.col("g"))
  }

  functionTest("char_length") {
    fn.char_length(fn.col("g"))
  }

  functionTest("character_length") {
    fn.character_length(fn.col("g"))
  }

  functionTest("chr") {
    fn.chr(fn.col("a"))
  }

  functionTest("contains") {
    fn.contains(fn.col("g"), fn.col("g"))
  }

  functionTest("elt") {
    fn.elt(fn.col("a"), fn.col("g"), fn.col("g"))
  }

  functionTest("find_in_set") {
    fn.find_in_set(fn.col("g"), fn.col("g"))
  }

  functionTest("like") {
    fn.like(fn.col("g"), fn.col("g"))
  }

  functionTest("like with escape") {
    fn.like(fn.col("g"), fn.col("g"), lit('/'))
  }

  functionTest("ilike") {
    fn.ilike(fn.col("g"), fn.col("g"))
  }

  functionTest("ilike with escape") {
    fn.ilike(fn.col("g"), fn.col("g"), lit('/'))
  }

  functionTest("lcase") {
    fn.lcase(fn.col("g"))
  }

  functionTest("ucase") {
    fn.ucase(fn.col("g"))
  }

  functionTest("left") {
    fn.left(fn.col("g"), fn.col("g"))
  }

  functionTest("right") {
    fn.right(fn.col("g"), fn.col("g"))
  }

  functionTest("array_agg") {
    fn.array_agg(fn.col("a"))
  }

  functionTest("array_size") {
    fn.array_size(fn.col("e"))
  }

  functionTest("cardinality") {
    fn.cardinality(fn.col("f"))
  }

  functionTest("count_min_sketch") {
    fn.count_min_sketch(fn.col("a"), fn.lit(0.1), fn.lit(0.95), fn.lit(11))
  }

  functionTest("named_struct") {
    fn.named_struct(fn.lit("x"), fn.col("a"), fn.lit("y"), fn.col("id"))
  }

  functionTest("json_array_length") {
    fn.json_array_length(fn.col("g"))
  }

  functionTest("json_object_keys") {
    fn.json_object_keys(fn.col("g"))
  }

  functionTest("mask with specific upperChar lowerChar digitChar otherChar") {
    fn.mask(fn.col("g"), fn.lit('X'), fn.lit('x'), fn.lit('n'), fn.lit('*'))
  }

  functionTest("mask with specific upperChar lowerChar digitChar") {
    fn.mask(fn.col("g"), fn.lit('X'), fn.lit('x'), fn.lit('n'))
  }

  functionTest("mask with specific upperChar lowerChar") {
    fn.mask(fn.col("g"), fn.lit('X'), fn.lit('x'))
  }

  functionTest("mask with specific upperChar") {
    fn.mask(fn.col("g"), fn.lit('X'))
  }

  functionTest("mask") {
    fn.mask(fn.col("g"))
  }

  functionTest("aes_encrypt with mode padding iv aad") {
    fn.aes_encrypt(
      fn.col("g"),
      fn.col("g"),
      fn.col("g"),
      fn.col("g"),
      fn.lit(Array(67.toByte, 68.toByte, 69.toByte)),
      fn.col("g"))
  }

  functionTest("aes_encrypt with mode padding iv") {
    fn.aes_encrypt(
      fn.col("g"),
      fn.col("g"),
      fn.col("g"),
      fn.col("g"),
      fn.lit(Array(67.toByte, 68.toByte, 69.toByte)))
  }

  functionTest("aes_encrypt with mode padding") {
    fn.aes_encrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("aes_encrypt with mode") {
    fn.aes_encrypt(fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("aes_encrypt") {
    fn.aes_encrypt(fn.col("g"), fn.col("g"))
  }

  functionTest("aes_decrypt with mode padding aad") {
    fn.aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("aes_decrypt with mode padding") {
    fn.aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("aes_decrypt with mode") {
    fn.aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("aes_decrypt") {
    fn.aes_decrypt(fn.col("g"), fn.col("g"))
  }

  functionTest("try_aes_decrypt with mode padding aad") {
    fn.try_aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("try_aes_decrypt with mode padding") {
    fn.try_aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("try_aes_decrypt with mode") {
    fn.try_aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("try_aes_decrypt") {
    fn.try_aes_decrypt(fn.col("g"), fn.col("g"))
  }

  functionTest("sha") {
    fn.sha(fn.col("g"))
  }

  functionTest("input_file_block_length") {
    fn.input_file_block_length()
  }

  functionTest("input_file_block_start") {
    fn.input_file_block_start()
  }

  functionTest("reflect") {
    fn.reflect(lit("java.util.UUID"), lit("fromString"), fn.col("g"))
  }

  functionTest("java_method") {
    fn.java_method(lit("java.util.UUID"), lit("fromString"), fn.col("g"))
  }

  functionTest("typeof") {
    fn.typeof(fn.col("g"))
  }

  functionTest("stack") {
    fn.stack(lit(2), fn.col("g"), fn.col("g"), fn.col("g"))
  }

  functionTest("random with seed") {
    fn.random(lit(1))
  }

  functionTest("call_function") {
    fn.call_function("lower", fn.col("g"))
  }

  test("hll_sketch_agg with column lgConfigK") {
    binary.select(fn.hll_sketch_agg(fn.col("bytes"), lit(0)))
  }

  test("hll_sketch_agg with column lgConfigK_int") {
    binary.select(fn.hll_sketch_agg(fn.col("bytes"), 0))
  }

  test("hll_sketch_agg with columnName lgConfigK_int") {
    binary.select(fn.hll_sketch_agg("bytes", 0))
  }

  test("hll_sketch_agg") {
    binary.select(fn.hll_sketch_agg(fn.col("bytes")))
  }

  test("hll_sketch_agg with columnName") {
    binary.select(fn.hll_sketch_agg("bytes"))
  }

  test("hll_union_agg with column allowDifferentLgConfigK") {
    binary.select(fn.hll_union_agg(fn.col("bytes"), lit(false)))
  }

  test("hll_union_agg with column allowDifferentLgConfigK_boolean") {
    binary.select(fn.hll_union_agg(fn.col("bytes"), false))
  }

  test("hll_union_agg with columnName allowDifferentLgConfigK_boolean") {
    binary.select(fn.hll_union_agg("bytes", false))
  }

  test("hll_union_agg") {
    binary.select(fn.hll_union_agg(fn.col("bytes")))
  }

  test("hll_union_agg with columnName") {
    binary.select(fn.hll_union_agg("bytes"))
  }

  test("groupby agg") {
    simple
      .groupBy(Column("id"))
      .agg(
        "a" -> "max",
        "b" -> "stddev",
        "b" -> "std",
        "b" -> "mean",
        "b" -> "average",
        "b" -> "avg",
        "*" -> "size",
        "a" -> "count")
  }

  test("groupby agg string") {
    simple
      .groupBy("id", "b")
      .agg("a" -> "max", "a" -> "count")
  }

  test("groupby agg columns") {
    simple
      .groupBy(Column("id"))
      .agg(functions.max("a"), functions.sum("b"))
  }

  test("groupby max") {
    simple
      .groupBy(Column("id"))
      .max("a", "b")
  }

  test("groupby min") {
    simple
      .groupBy(Column("id"))
      .min("a", "b")
  }

  test("groupby mean") {
    simple
      .groupBy(Column("id"))
      .mean("a", "b")
  }

  test("groupby avg") {
    simple
      .groupBy(Column("id"))
      .avg("a", "b")
  }

  test("groupby sum") {
    simple
      .groupBy(Column("id"))
      .sum("a", "b")
  }

  test("groupby count") {
    simple
      .groupBy(Column("id"))
      .count()
  }

  test("rollup column") {
    simple.rollup(Column("a"), Column("b")).count()
  }

  test("cube column") {
    simple.cube(Column("a"), Column("b")).count()
  }

  test("rollup string") {
    simple.rollup("a", "b").count()
  }

  test("cube string") {
    simple.cube("a", "b").count()
  }

  test("grouping and grouping_id") {
    simple
      .cube("a", "b")
      .agg(fn.grouping("a"), fn.grouping("b"), fn.grouping_id("a", "b"))
  }

  test("pivot") {
    simple.groupBy(Column("id")).pivot("a", Seq(1, 2, 3)).agg(functions.count(Column("b")))
  }

  test("pivot without column values") {
    simple.groupBy(Column("id")).pivot("a").agg(functions.count(Column("b")))
  }

  test("width_bucket") {
    simple.select(fn.width_bucket(fn.col("b"), fn.col("b"), fn.col("b"), fn.col("a")))
  }

  test("test broadcast") {
    left.join(fn.broadcast(right), "id")
  }

  test("function lit") {
    simple.select(
      fn.lit(fn.col("id")),
      fn.lit('id),
      fn.lit(true),
      fn.lit(68.toByte),
      fn.lit(9872.toShort),
      fn.lit(-8726532),
      fn.lit(7834609328726532L),
      fn.lit(Math.E),
      fn.lit(-0.8f),
      fn.lit(BigDecimal(8997620, 5)),
      fn.lit(BigDecimal(898897667231L, 7).bigDecimal),
      fn.lit("connect!"),
      fn.lit('T'),
      fn.lit(Array.tabulate(10)(i => ('A' + i).toChar)),
      fn.lit(Array.tabulate(23)(i => (i + 120).toByte)),
      fn.lit(mutable.WrappedArray.make(Array[Byte](8.toByte, 6.toByte))),
      fn.lit(null),
      fn.lit(java.time.LocalDate.of(2020, 10, 10)),
      fn.lit(Decimal.apply(BigDecimal(8997620, 6))),
      fn.lit(java.time.Instant.ofEpochMilli(1677155519808L)),
      fn.lit(new java.sql.Timestamp(12345L)),
      fn.lit(java.time.LocalDateTime.of(2023, 2, 23, 20, 36)),
      fn.lit(java.sql.Date.valueOf("2023-02-23")),
      fn.lit(java.time.Duration.ofSeconds(200L)),
      fn.lit(java.time.Period.ofDays(100)),
      fn.lit(new CalendarInterval(2, 20, 100L)))
  }

  test("function lit array") {
    simple.select(
      fn.lit(Array.emptyDoubleArray),
      fn.lit(Array(Array(1), Array(2), Array(3))),
      fn.lit(Array(Array(Array(1)), Array(Array(2)), Array(Array(3)))),
      fn.lit(Array(true, false)),
      fn.lit(Array(67.toByte, 68.toByte, 69.toByte)),
      fn.lit(Array(9872.toShort, 9873.toShort, 9874.toShort)),
      fn.lit(Array(-8726532, 8726532, -8726533)),
      fn.lit(Array(7834609328726531L, 7834609328726532L, 7834609328726533L)),
      fn.lit(Array(Math.E, 1.toDouble, 2.toDouble)),
      fn.lit(Array(-0.8f, -0.7f, -0.9f)),
      fn.lit(Array(BigDecimal(8997620, 5), BigDecimal(8997621, 5))),
      fn.lit(
        Array(BigDecimal(898897667231L, 7).bigDecimal, BigDecimal(898897667231L, 7).bigDecimal)),
      fn.lit(Array("connect!", "disconnect!")),
      fn.lit(Array('T', 'F')),
      fn.lit(
        Array(
          Array.tabulate(10)(i => ('A' + i).toChar),
          Array.tabulate(10)(i => ('B' + i).toChar))),
      fn.lit(Array(java.time.LocalDate.of(2020, 10, 10), java.time.LocalDate.of(2020, 10, 11))),
      fn.lit(
        Array(
          java.time.Instant.ofEpochMilli(1677155519808L),
          java.time.Instant.ofEpochMilli(1677155519809L))),
      fn.lit(Array(new java.sql.Timestamp(12345L), new java.sql.Timestamp(23456L))),
      fn.lit(
        Array(
          java.time.LocalDateTime.of(2023, 2, 23, 20, 36),
          java.time.LocalDateTime.of(2023, 2, 23, 21, 36))),
      fn.lit(Array(java.sql.Date.valueOf("2023-02-23"), java.sql.Date.valueOf("2023-03-01"))),
      fn.lit(Array(java.time.Duration.ofSeconds(100L), java.time.Duration.ofSeconds(200L))),
      fn.lit(Array(java.time.Period.ofDays(100), java.time.Period.ofDays(200))),
      fn.lit(Array(new CalendarInterval(2, 20, 100L), new CalendarInterval(2, 21, 200L))))
  }

  test("function typedLit") {
    simple.select(
      fn.typedLit(fn.col("id")),
      fn.typedLit('id),
      fn.typedLit(1),
      fn.typedLit[String](null),
      fn.typedLit(true),
      fn.typedLit(68.toByte),
      fn.typedLit(9872.toShort),
      fn.typedLit(-8726532),
      fn.typedLit(7834609328726532L),
      fn.typedLit(Math.E),
      fn.typedLit(-0.8f),
      fn.typedLit(BigDecimal(8997620, 5)),
      fn.typedLit(BigDecimal(898897667231L, 7).bigDecimal),
      fn.typedLit("connect!"),
      fn.typedLit('T'),
      fn.typedLit(Array.tabulate(10)(i => ('A' + i).toChar)),
      fn.typedLit(Array.tabulate(23)(i => (i + 120).toByte)),
      fn.typedLit(mutable.WrappedArray.make(Array[Byte](8.toByte, 6.toByte))),
      fn.typedLit(null),
      fn.typedLit(java.time.LocalDate.of(2020, 10, 10)),
      fn.typedLit(Decimal.apply(BigDecimal(8997620, 6))),
      fn.typedLit(java.time.Instant.ofEpochMilli(1677155519808L)),
      fn.typedLit(new java.sql.Timestamp(12345L)),
      fn.typedLit(java.time.LocalDateTime.of(2023, 2, 23, 20, 36)),
      fn.typedLit(java.sql.Date.valueOf("2023-02-23")),
      fn.typedLit(java.time.Duration.ofSeconds(200L)),
      fn.typedLit(java.time.Period.ofDays(100)),
      fn.typedLit(new CalendarInterval(2, 20, 100L)),

      // Handle parameterized scala types e.g.: List, Seq and Map.
      fn.typedLit(Some(1)),
      fn.typedLit(Array(1, 2, 3)),
      fn.typedLit(Seq(1, 2, 3)),
      fn.typedLit(Map("a" -> 1, "b" -> 2)),
      fn.typedLit(("a", 2, 1.0)),
      fn.typedLit[Option[Int]](None),
      fn.typedLit[Array[Option[Int]]](Array(Some(1))),
      fn.typedlit[Map[Int, Option[Int]]](Map(1 -> None)),
      fn.typedlit[mutable.Map[Int, Option[Int]]](mutable.Map(1 -> None)),
      fn.typedlit[collection.immutable.Map[Int, Option[Int]]](
        collection.immutable.Map(1 -> None)),
      fn.typedLit(Seq(Seq(1, 2, 3), Seq(4, 5, 6), Seq(7, 8, 9))),
      fn.typedLit(Seq(Map("a" -> 1, "b" -> 2), Map("a" -> 3, "b" -> 4), Map("a" -> 5, "b" -> 6))),
      fn.typedLit(Map(1 -> Map("a" -> 1, "b" -> 2), 2 -> Map("a" -> 3, "b" -> 4))),
      fn.typedLit((Seq(1, 2, 3), Map("a" -> 1, "b" -> 2), ("a", Map(1 -> "a", 2 -> "b")))))
  }

  /* Window API */
  test("window") {
    simple.select(
      fn.min("id").over(Window.partitionBy(Column("a"), Column("b"))),
      fn.min("id").over(Window.partitionBy("a", "b")),
      fn.min("id").over(Window.orderBy(Column("a"), Column("b"))),
      fn.min("id").over(Window.orderBy("a", "b")),
      fn.min("id").over(Window.orderBy("a").rowsBetween(2L, 3L)),
      fn.min("id").over(Window.orderBy("a").rangeBetween(2L, 3L)),
      fn.count(Column("id")).over())
  }

  /* Extensions */
  test("relation extension") {
    val input = proto.ExamplePluginRelation
      .newBuilder()
      .setInput(simple.plan.getRoot)
      .build()
    session.newDataFrame(com.google.protobuf.Any.pack(input))
  }

  test("expression extension") {
    val extension = proto.ExamplePluginExpression
      .newBuilder()
      .setChild(
        proto.Expression
          .newBuilder()
          .setUnresolvedAttribute(proto.Expression.UnresolvedAttribute
            .newBuilder()
            .setUnparsedIdentifier("id")))
      .setCustomField("abc")
      .build()
    simple.select(Column(com.google.protobuf.Any.pack(extension)))
  }

  test("crosstab") {
    simple.stat.crosstab("a", "b")
  }

  test("freqItems") {
    simple.stat.freqItems(Array("id", "a"), 0.1)
  }

  test("sampleBy") {
    simple.stat.sampleBy("id", Map(0 -> 0.1, 1 -> 0.2), 0L)
  }

  test("drop") {
    simple.na.drop(5, Seq("id", "a"))
  }

  test("fill") {
    simple.na.fill(8L, Seq("id"))
  }

  test("replace") {
    simple.na.replace[Long]("id", Map(1L -> 8L))
  }

  /* Reader API  */
  test("table API with options") {
    session.read.options(Map("p1" -> "v1", "p2" -> "v2")).table("tempdb.myTable")
  }

  /* Stream Reader API  */
  test("streaming table API with options") {
    session.readStream.options(Map("p1" -> "v1", "p2" -> "v2")).table("tempdb.myStreamingTable")
  }

  /* Avro functions */
  test("from_avro with options") {
    binary.select(
      avroFn.from_avro(
        fn.col("bytes"),
        """{"type": "int", "name": "id"}""",
        Map("mode" -> "FAILFAST", "compression" -> "zstandard").asJava))
  }

  test("from_avro without options") {
    binary.select(avroFn.from_avro(fn.col("bytes"), """{"type": "string", "name": "name"}"""))
  }

  test("to_avro with schema") {
    simple.select(avroFn.to_avro(fn.col("a"), """{"type": "int", "name": "id"}"""))
  }

  test("to_avro without schema") {
    simple.select(avroFn.to_avro(fn.col("id")))
  }

  /* Protobuf functions */
  // scalastyle:off line.size.limit
  // If `common.desc` needs to be updated, execute the following command to regenerate it:
  //  1. cd connector/connect/common/src/main/protobuf/spark/connect
  //  2. protoc --include_imports --descriptor_set_out=../../../../test/resources/protobuf-tests/common.desc common.proto
  // scalastyle:on line.size.limit
  private val testDescFilePath: String = s"${IntegrationTestUtils.sparkHome}/connector/" +
    "connect/common/src/test/resources/protobuf-tests/common.desc"

  test("from_protobuf messageClassName") {
    binary.select(pbFn.from_protobuf(fn.col("bytes"), classOf[StorageLevel].getName))
  }

  test("from_protobuf messageClassName options") {
    binary.select(
      pbFn.from_protobuf(
        fn.col("bytes"),
        classOf[StorageLevel].getName,
        Map("recursive.fields.max.depth" -> "2").asJava))
  }

  test("from_protobuf messageClassName descFilePath options") {
    binary.select(
      pbFn.from_protobuf(
        fn.col("bytes"),
        "StorageLevel",
        testDescFilePath,
        Map("recursive.fields.max.depth" -> "2").asJava))
  }

  test("from_protobuf messageClassName descFilePath") {
    binary.select(pbFn.from_protobuf(fn.col("bytes"), "StorageLevel", testDescFilePath))
  }

  test("to_protobuf messageClassName") {
    binary.select(pbFn.to_protobuf(fn.col("bytes"), classOf[StorageLevel].getName))
  }

  test("to_protobuf messageClassName options") {
    binary.select(
      pbFn.to_protobuf(
        fn.col("bytes"),
        classOf[StorageLevel].getName,
        Map("recursive.fields.max.depth" -> "2").asJava))
  }

  test("to_protobuf messageClassName descFilePath options") {
    binary.select(
      pbFn.to_protobuf(
        fn.col("bytes"),
        "StorageLevel",
        testDescFilePath,
        Map("recursive.fields.max.depth" -> "2").asJava))
  }

  test("to_protobuf messageClassName descFilePath") {
    binary.select(pbFn.to_protobuf(fn.col("bytes"), "StorageLevel", testDescFilePath))
  }
}
